package com.jplus.plugin.rpc.nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.jplus.plugin.rpc.process.NIOProcess;

public abstract class NIO extends Thread {
	public static enum NIORole {
		CLIENT, SERVER;
	}

	final Logger log = LoggerFactory.getLogger(NIO.class);
	protected Selector selector;
	protected NIOProcess process;
	protected ExecutorService threadPool;

	@Override
	public void run() {
		try {
			while (!Thread.interrupted()) {
				int events = selector.select(1000);// 阻塞1s
				if (events > 0) {
					Set<SelectionKey> selectionKeys = selector.selectedKeys();
					Iterator<SelectionKey> iterator = selectionKeys.iterator();
					while (iterator.hasNext()) {
						SelectionKey sk = iterator.next();
						iterator.remove();
						handleKey(sk);
					}
				}
			}
		} catch (IOException e) {
			log.error("NIO ERROR:", e);
		}
	}

	private void handleKey(SelectionKey sk) {
		try {
			if (sk.isValid() && sk.isConnectable()) {// 客户端通道
				SocketChannel sc = (SocketChannel) sk.channel();
				sc.configureBlocking(false);
				sc.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);// 开始读写
				log.info("Client:is ready to write...");
			}
			if (sk.isValid() && sk.isAcceptable()) {
				ServerSocketChannel ss = (ServerSocketChannel) sk.channel();// 服务端通道
				SocketChannel sc = ss.accept();
				sc.configureBlocking(false);
				sc.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);// 开始读写
				log.info("Server:is a new client link:" + sc.getRemoteAddress());
			}
			if (sk.isReadable()) {
				SocketChannel sc = (SocketChannel) sk.channel();
				ByteBuffer buffer = ByteBuffer.allocate(1024);
				sc.read(buffer);
				process.parseRead(buffer);
			}
			if (sk.isWritable()) {
				SocketChannel sc = (SocketChannel) sk.channel();
				ByteBuffer buffer = process.parseWrite();
				if (buffer != null)
					sc.write(buffer);
			}
		} catch (IOException e) {
			log.error("Handle ERROR:", e);
			sk.cancel();
		}
	}
}
